package org.zhadoop.stream.cql.semanticanalyzer;

import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zhadoop.stream.api.streams.Column;
import org.zhadoop.stream.api.streams.Schema;
import org.zhadoop.stream.cql.exception.SemanticAnalyzerException;
import org.zhadoop.stream.cql.semanticanalyzer.analyzecontext.AnalyzeContext;
import org.zhadoop.stream.cql.semanticanalyzer.parser.context.ParseContext;
import org.zhadoop.streaming.exception.ErrorCode;

import com.google.common.collect.Sets;

/**
 * 语义分析的基础类
 * @author zwd
 *
 */
public abstract class BaseAnalyzer implements SemanticAnalyzer {
	
	private static final Logger LOG = LoggerFactory.getLogger(BaseAnalyzer.class);
	
	/**
	 * 默认的列名称
	 * 当sql没有显示的指定列名称的时候，就需要对列进行自动命名，这个时候用到
	 */
	private static final String DEFAULT_COLUMN_PRIFIX = "x_col_";
	
	/**
	 * 对列进行重命名时的后缀
	 */
	private static final String RENAME_COLUMN_POSTFIX = "_";
	
	private List<Schema> allSchemas = null;
	
	private ParseContext parseContext;
	
	public BaseAnalyzer(ParseContext parseContext) throws SemanticAnalyzerException{
		this.parseContext = parseContext;
	}
	
	/**
	 * java 初始化的顺序
	 * 父类--静态变量
	 * 父类--静态初始化块
	 * 子类--静态化变量
	 * 子类--静态初始化块
	 * 父类--变量
	 * 父类--初始化块
	 * 父类--构造器
	 * 子类--变量
	 * 子类--初始化块
	 * 子类--构造器
	 */
	@Override
	public void init(List<Schema> schemas) throws SemanticAnalyzerException {
		/**
		 * 由于java 先初始化父类的一切，再初始化子类的变量和构造器
		 * 所以这里的这些方法必须放在init 方法中
		 * 否则就会发生子类的变量被重新初始化，导致空指针的问题
		 */
		createAnalyzeContext();
		getAnalyzeContext().setParseContext(parseContext);
		getAnalyzeContext().validateParseContext();
	}
	

	/**
	 * 获取语义分析结果对象
	 * @return
	 */
	protected abstract AnalyzeContext getAnalyzeContext();

	/**
	 * 创建analyze语义分析结果保存对象
	 */
	protected abstract void createAnalyzeContext();
	
	public List<Schema> getAllSchemas() {
		return allSchemas;
	}
	
	protected Schema getSchemaByName(String name)
		throws SemanticAnalyzerException{
		for(Schema schema : allSchemas){
			if(compareSchema(name, schema)){
				return schema;
			}
		}
		
		SemanticAnalyzerException exception =
				new SemanticAnalyzerException(ErrorCode.SEMANTICANALYZE_NOFOUND_STREAM, name);
		LOG.error("Can't find schema by stream name.", exception);
		throw exception;
	}
	
	/**
	 * 通过名称获取schema
	 * @param name
	 * @param schemas
	 * @return
	 * @throws SemanticAnalyzerException
	 */
	public static Schema getSchemaByName(String name,List<Schema> schemas)
	throws SemanticAnalyzerException{
		for(Schema schema : schemas){
			if(compareSchema(name,schema)){
				return schema;
			}
		}
		SemanticAnalyzerException exception = 
				new SemanticAnalyzerException(ErrorCode.SEMANTICANALYZE_NOFOUND_STREAM,name);
		LOG.error("Can't find schema by stream name.", exception);
		throw exception;
	}
	
	/**
	 * 修改schema 中每个属性的schema名称
	 * 统一schema 的名称和列名称，均为小写
	 * @param schemas
	 */
	public static void setSchemaNameInAttributes(List<Schema> schemas){
		for(int i = 0; i < schemas.size(); i++){
			Schema s = schemas.get(i);
			for(Column attr : s.getCols()){
				attr.setName(attr.getName().toLowerCase(Locale.US));
			}
		}
	}
	
	/**
	 * 检查schema 是否存在
	 * @param name
	 * @return
	 */
	protected boolean checkSchemaExists(String name){
		for(Schema schema : allSchemas){
			if(compareSchema(name, schema)){
				return true;
			}
		}
		return false;
	}
	
	protected void checkSchemas(List<Schema> schemasInCQL)
	throws SemanticAnalyzerException{
		Set<String> names = Sets.newHashSet();
		for(Schema s : schemasInCQL){
			/**
			 * schema 名称不能为空
			 * 但是流的别名可能为空
			 */
			if(names.contains(s.getId())){
				SemanticAnalyzerException exception 
					= new SemanticAnalyzerException(ErrorCode.SEMANTICANALYZE_EXISTS_STREAM, s.getId());
				LOG.error("Stream {} already exists.", s.getId(),exception);
				throw exception;
			}
		}
	}

	/**
	 * 对于已经存在的列，重名的列，自动重命名
	 * @param schema
	 * @param colName
	 * @return
	 */
	protected String renameNewName(Schema schema,String colName){
		/**
		 * 这个时候，schema和其中的列一定不为空
		 * 因为在进入这个方法之前，肯定已经检查过是否存在同名列了
		 * 
		 * 并且由于前面已经存在同名列了，所以这个循环检查计数器要从1开始
		 */
		int count = 1;
		for(Column attr :schema.getCols()){
			if(attr.getName().startsWith(colName + RENAME_COLUMN_POSTFIX)){
				count ++;
			}
		}
		return colName + RENAME_COLUMN_POSTFIX + count;
	}
	
	/**
	 * 对于通过表达式计算出来的列，需要创建新的列
	 * @param schema
	 * @return
	 */
	protected String createNewName(Schema schema){
		int count = 0;
		if(schema != null & schema.getCols() != null){
			for(Column attr : schema.getCols()){
				if(attr.getName().startsWith(DEFAULT_COLUMN_PRIFIX)){
					count ++;
				}
			}
		}
		return DEFAULT_COLUMN_PRIFIX + count;
	}
	
	/**
	 * 获取一个schema中所有属性
	 * @param schema
	 * @return
	 */
	public static List<Column> getAttributes(Schema schema){
		return schema.getCols();
	}
	
	/**
	 * 获取所有流的所有属性
	 * 适用于select *
	 * @param schemas
	 * @return
	 */
	protected static List<Column> getAllAttributes(List<Schema> schemas){
		List<Column> attrs = new ArrayList<Column>();
		for(Schema schema : schemas){
			attrs.addAll(getAttributes(schema));
		}
		return attrs;
	}
	
	private static boolean compareSchema(String name, Schema schema) {
		if(!StringUtils.isEmpty(schema.getId()) && schema.getId().equalsIgnoreCase(name)){
			return true;
		}
		
		if(!StringUtils.isEmpty(schema.getName()) && schema.getName().equalsIgnoreCase(name)){
			return true;
		}
		
		if(!StringUtils.isEmpty(schema.getStreamName()) && schema.getStreamName().equalsIgnoreCase(name)){
			return true;
		}
		return false;
	}
	
	/**
	 * 通过属性名获取属性
	 * 可能包含多个
	 * @param attrName
	 * @param schema
	 * @param schemas
	 * @return
	 */
	public static List<Column> getAttributeByName(String attrName,Schema schema,List<Schema> schemas){
		List<Column> allAttrs = null;
		if(schema == null){
			allAttrs = getAllAttributes(schemas);
		}else{
			allAttrs=getAttributes(schema);
		}
		return  getColumnByNameOrALias(allAttrs, attrName);
	}

	/**
	 * 通过某个属性名获取属性值
	 * 可能包含多个
	 * @param attrName
	 * @param schemas
	 * @return
	 */
	public static List<Column> getAttributeByName(String attrName,List<Schema> schemas){
		List<Column> allAttrs = getAllAttributes(schemas);
		return getColumnByNameOrALias(allAttrs, attrName);
	}
	
	
	private static List<Column> getColumnByNameOrALias(List<Column> columns, String nameOrAlias) {
		List<Column> res = new ArrayList<Column>();
		for(Column attr : columns){
			if(!StringUtils.isEmpty(attr.getName()) && attr.getName().equalsIgnoreCase(nameOrAlias)){
				res.add(attr);
				continue;
			}
			
			if(!StringUtils.isEmpty(attr.getAlias()) && attr.getAlias().equalsIgnoreCase(nameOrAlias)){
				res.add(attr);
				continue;
			}
		}
		return res;
	}
	
}
